package com.yunc.base.queue.disruptor;

import com.lmax.disruptor.BlockingWaitStrategy;
import com.lmax.disruptor.EventHandler;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.ProducerType;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.stereotype.Service;

import java.util.Set;
import java.util.concurrent.Executors;

@Service
public class DisruptorService implements DisposableBean, InitializingBean {

    private Disruptor<Event> disruptor;
    private static final int RING_BUFFER_SIZE = 1024 * 1024;

    @Override
    public void destroy() {
        disruptor.shutdown();
    }

    @Override
    public void afterPropertiesSet() {
        EventHandlerProcess process = new EventHandlerProcess();
        disruptor = new Disruptor<>(Event.FACTORY, RING_BUFFER_SIZE, Executors.defaultThreadFactory(), ProducerType.SINGLE,new BlockingWaitStrategy());
        // 手动添加handler
        Set<EventHandler> eventHandlers = process.scanHandlers();
        for (EventHandler handler : eventHandlers) {
            // if (filterHandler(handler)) {
            disruptor.handleEventsWith(handler);
        }
        disruptor.start();
    }

    public <T> void publishEvent(T event) {
        Event commEvent = new Event(event);
        final long seq = disruptor.getRingBuffer().next();
        Event userEvent = disruptor.get(seq);
        userEvent.setValue(commEvent.getValue());
        disruptor.getRingBuffer().publish(seq);
    }
}
